package com.burges.net.tableAPIAndSQL.time

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/15 
  * 描述     临时表通过TemporalTableFunction定义 基于Append-Only Table来定义
  */
object TemporalTablesDEMO {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		// 获取TableEnvironment
		val tStreamEnv: StreamTableEnvironment = TableEnvironment.getTableEnvironment(environment)
		//创建流式数据集
		val ds: DataStream[(Long, Int)] = environment.fromElements((512L, 1))
		//将DataStream数据集转换成Table
		val tempTable = ds.toTable(tStreamEnv, 't_id', 't_value', 't_proctime.proctime')
		//在TableEnvironment中注册表结构
		tStreamEnv.registerTable("tempTable", tempTable)
		//调用createTemporalTableFunction注册临时表，指定事件属性t_proctime及主键t_id
		val tempTableFunction = tempTable.createTemporalTableFunction('t_proctime', 't_id')
		//在TableEnvironment中注册tempTableFunction信息，然后在SQL中通过名称调用
		tStreamEnv.registerFunction("tempTable", tempTableFunction)
	}

}
